#include "ZKClient.h"
#include "Spark/Utils/Funcs.h"

const int ZKClient::EventType::CREATED = ZOO_CREATED_EVENT;
const int ZKClient::EventType::DELETED = ZOO_DELETED_EVENT;
const int ZKClient::EventType::CHANGED = ZOO_CHANGED_EVENT;
const int ZKClient::EventType::CHILD = ZOO_CHILD_EVENT;
const int ZKClient::EventType::SESSION = ZOO_SESSION_EVENT;
const int ZKClient::EventType::NOWATCHING = ZOO_NOTWATCHING_EVENT;

const int ZKClient::FlagType::EPHEMERAL = ZOO_EPHEMERAL;
const int ZKClient::FlagType::SEQUENCE = ZOO_SEQUENCE;
const int ZKClient::FlagType::CONTAINER = ZOO_CONTAINER;

const int ZKClient::StateType::EXPIRED_SESSION = ZOO_EXPIRED_SESSION_STATE;
const int ZKClient::StateType::AUTH_FAILED = ZOO_AUTH_FAILED_STATE;
const int ZKClient::StateType::CONNECTING = ZOO_CONNECTING_STATE;
const int ZKClient::StateType::ASSOCIATING = ZOO_ASSOCIATING_STATE;
const int ZKClient::StateType::CONNECTED = ZOO_CONNECTED_STATE;
const int ZKClient::StateType::READONLY = ZOO_READONLY_STATE;
const int ZKClient::StateType::NOTCONNECTED = ZOO_NOTCONNECTED_STATE;

ZKClient::ZKClient()
{
}

ZKClient::~ZKClient()
{
    if (handle_)
    {
        Close();
    }
}

int ZKClient::Init(const std::string& hosts,
                   int timeout,
                   SharedWatherFn wcb,
                   LogCallback lcb)
{
    using namespace std::placeholders;
    if (handle_)
    {
        return true;
    }
    hosts_ = hosts;
    timeout_ = timeout;
    watcherCallback_ = std::bind(wcb, _1, _2, _3, shared_from_this());
    logCallback_ = std::move(lcb);
    handle_ = zookeeper_init2(hosts.c_str(),
                              &ZKClient::watcherCallback,
                              timeout_,
                              nullptr,
                              this,
                              0,
                              lcb);
    return handle_ ? 0 : -1;
}

int ZKClient::SetHosts(const std::string& hosts)
{
    int ret = zoo_set_servers(handle_, hosts.c_str());
    if (ret == 0)
    {
        hosts_ = hosts;
    }
    return ret;
}

int ZKClient::Create(const std::string& path,
                     const std::string& val,
                     std::string& newPath,
                     const ACL_vector* acl,
                     int flags)
{
    return zoo_create(handle_,
                      path.c_str(),
                      val.c_str(),
                      static_cast<int>(val.size()),
                      acl,
                      flags,
                      &newPath[0],
                      static_cast<int>(newPath.size()));
}

int ZKClient::Exists(const std::string& path, bool watch, Stat* stat)
{
    return zoo_exists(handle_, path.c_str(), watch, stat);
}

int ZKClient::Delete(const std::string& path, int version)
{
    return zoo_delete(handle_, path.c_str(), version);
}

int ZKClient::Get(const std::string& path,
                  std::string& val,
                  bool watch,
                  Stat* stat)
{
    int len = static_cast<int>(val.size());
    int ret = zoo_get(handle_, path.c_str(), watch, &val[0], &len, stat);
    if (ret == ZOK)
    {
        val.resize(len);
    }
    return ret;
}

int ZKClient::GetConfig(std::string& val, bool watch, Stat* stat)
{
    return Get(ZOO_CONFIG_NODE, val, watch, stat);
}

int ZKClient::Set(const std::string& path,
                  const std::string& val,
                  int version,
                  Stat* stat)
{
    return zoo_set2(handle_,
                    path.c_str(),
                    val.c_str(),
                    static_cast<int>(val.size()),
                    version,
                    stat);
}

int ZKClient::GetChildren(const std::string& path,
                          std::vector<std::string>& val,
                          bool watch,
                          Stat* stat)
{
    String_vector strings;
    Stat tmp;
    if (stat == nullptr)
    {
        stat = &tmp;
    }
    int ret = zoo_get_children2(handle_, path.c_str(), watch, &strings, stat);
    if (ret == ZOK)
    {
        for (int i = 0; i < strings.count; ++i)
        {
            val.emplace_back(strings.data[i]);
        }
        deallocate_String_vector(&strings);
    }
    return ret;
}

int ZKClient::Close()
{
    watcherCallback_ = nullptr;
    int ret = ZOK;
    if (handle_)
    {
        ret = zookeeper_close(handle_);
        handle_ = nullptr;
    }
    return ret;
}

int ZKClient::GetState()
{
    return zoo_state(handle_);
}

std::string ZKClient::GetCurrentServer()
{
    const char* ret = zoo_get_current_server(handle_);
    return ret == nullptr ? "" : ret;
}

int ZKClient::Reconnect()
{
    if (handle_)
    {
        zookeeper_close(handle_);
    }
    handle_ = zookeeper_init2(hosts_.c_str(),
                              &ZKClient::watcherCallback,
                              timeout_,
                              nullptr,
                              this,
                              0,
                              logCallback_);
    return handle_ ? 0 : -1;
}

void ZKClient::watcherCallback(zhandle_t* zh,
                               int type,
                               int stat,
                               const char* path,
                               void* watcherCtx)
{
    ZKClient* cli = static_cast<ZKClient*>(watcherCtx);
    ASSERT_(cli != nullptr);
    cli->watcherCallback_(type, stat, path);
}
